package cn._51doit.flink.day06;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.io.RandomAccessFile;
import java.util.Iterator;

/**
 * 使用OperatorState要实现CheckpointedFunction接口，然后重新两个方法
 * 1.在initializeState方法中初始化或恢复状态（初始化时只执行一次，在open方法执行之前）
 * 2.在snapshotState方法中，将最新的数据保存到OperatorState中（每一次checkpoint执行一次）
 */

public class MyAtLeastOnceSource extends RichParallelSourceFunction<String> implements CheckpointedFunction {

    private String path;

    public MyAtLeastOnceSource(String path) {
        this.path = path;
    }

    private boolean flag = true;
    private long offset = 0;

    private transient ListState<Long> offsetState;

    //OperatorState要在initializeState方法中初始化
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        //System.out.println("initializeState method invoked ... subtask : " + getRuntimeContext().getIndexOfThisSubtask());
        ListStateDescriptor<Long> stateDescriptor = new ListStateDescriptor<>("offset-state", Long.class);
        offsetState = context.getOperatorStateStore().getListState(stateDescriptor);
        //将OperatorState中的偏移量取出来
        if (context.isRestored()) {
            Iterator<Long> iterator = offsetState.get().iterator();
            if (iterator.hasNext()) {
                offset = iterator.next();
            }
        }
    }


    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        System.out.println("run method invoked ... subtask : " + getRuntimeContext().getIndexOfThisSubtask());
        int index = getRuntimeContext().getIndexOfThisSubtask();
        RandomAccessFile file = new RandomAccessFile( path +  "/" + index + ".txt", "r");
        //从指定的偏移量开始读取数据
        file.seek(offset);

        while (flag) {
            String line = file.readLine();
            if (line != null) {
                line = new String(line.getBytes("ISO-8859-1"), "UTF-8");
                synchronized (ctx.getCheckpointLock()) {
                    ctx.collect(line);
                    //获取文件最新的偏移量
                    offset = file.getFilePointer();
                }
            } else {
                //没有数据，睡觉
                Thread.sleep(1000);
            }
        }

    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        //System.out.println("snapshotState method invoked ... subtask : " + getRuntimeContext().getIndexOfThisSubtask());
        offsetState.clear(); //清除原来的偏移量
        offsetState.add(offset); //将最新的偏移量添加到OperatorState
    }

    @Override
    public void cancel() {
        //System.out.println("cancel method invoked ... subtask : " + getRuntimeContext().getIndexOfThisSubtask());
        flag = false;
    }


}
